package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author Felix
 * @date 2022/11/1
 * 维度数据处理
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //处理主流业务数据
    //{"database":"gmall0530","table":"base_trademark","type":"insert","ts":1667285913,"xid":24766,"commit":true,"data":{"id":12,"tm_name":"四星","logo_url":"fdsaf"}}
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取操作的业务数据库表表名
        String tableName = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = broadcastState.get(tableName);
        if(tableProcess != null){
            //如果获取到的配置对象不为null，说明处理的是维度数据
            //将维度数据继续向下游传递 {"tm_name":"六星","logo_url":"fdsaf","id":12}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //过滤掉不需要向下游传递的属性  {"tm_name":"六星","id":12}
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);
            //在传递维度数据前，补充输出目的地  {"tm_name":"八星","sink_table":"dim_base_trademark","id":12}
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table",sinkTable);
            //补充操作的类型
            dataJsonObj.put("type",jsonObj.getString("type"));

            out.collect(dataJsonObj);
        }
    }

    //过滤掉不需要传递的属性
    // dataJsonObj：{"tm_name":"六星","logo_url":"fdsaf","id":12}
    // sinkColumns：id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));
    }

    //处理广播流配置数据
    //op:r {"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667268460352,"snapshot":"false","db":"gmall0530_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1667268460352,"transaction":null}
    //op:c {"before":null,"after":{"source_table":"a","sink_table":"dim_a","sink_columns":"aa,bb,cc","sink_pk":"aa","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667269006000,"snapshot":"false","db":"gmall0530_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000014","pos":385,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1667269005235,"transaction":null}
    //op:u {"before":{"source_table":"a","sink_table":"dim_a","sink_columns":"aa,bb,cc","sink_pk":"aa","sink_extend":null},"after":{"source_table":"a","sink_table":"dim_a","sink_columns":"aa,bb,cc,dd","sink_pk":"aa","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667269054000,"snapshot":"false","db":"gmall0530_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000014","pos":707,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1667269053202,"transaction":null}
    //op:d {"before":{"source_table":"a","sink_table":"dim_a","sink_columns":"aa,bb,cc,dd","sink_pk":"aa","sink_extend":null},"after":null,"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1667269077000,"snapshot":"false","db":"gmall0530_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000014","pos":1058,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1667269076036,"transaction":null}
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //为了操作方便，将jsonStr转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取对配置表的操作类型
        String op = jsonObj.getString("op");
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //判断对配置表是否进行了删除操作
        if("d".equals(op)){
            //如果对配置表做了删除操作，需要将对应的配置信息从广播状态中删除掉
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            String sourceTable = before.getSourceTable();
            broadcastState.remove(sourceTable);
        }else{
            //如果对配置表做了删除以外的其它操作，需要将最新的配置信息放到广播状态中
            TableProcess after = jsonObj.getObject("after", TableProcess.class);
            //获取业务系统中维度表表名
            String sourceTable = after.getSourceTable();
            //获取phoenix表名
            String sinkTable = after.getSinkTable();
            //获取表中字段
            String sinkColumns = after.getSinkColumns();
            //获取建表主键
            String sinkPk = after.getSinkPk();
            //获取建表扩展
            String sinkExtend = after.getSinkExtend();
            //提前将Phoenix中的维度表创建出来
            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
            //将对应的配置信息放到广播状态中
            broadcastState.put(sourceTable,after);
        }
    }

    //创建维度表
    private void checkTable(String tableName, String sinkColumns, String pk, String ext) {
        if(pk == null){
            pk = "id";
        }
        if(ext == null){
            ext = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.PHOENIX_SCHEMA +"."+tableName+"(");
        //将表中字段进行分隔
        String[] fieldArr = sinkColumns.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            //判断是否为主键字段
            if(pk.equals(field)){
                createSql.append(field + " varchar primary key");
            }else{
                createSql.append(field + " varchar");
            }
            //除了最后一个字段之外，其它字段在拼接的时候都需要加逗号
            if(i < fieldArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(")" + ext);
        System.out.println("在phoenix中执行的建表语句:" + createSql);

        PhoenixUtil.executeSql(createSql.toString());
    }
}